package com.orion.service;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;

/**
 * 单节点 a,b随便启动一个来测试
 * @author Administrator
 * @date 2021/12/2
 */
//@Component
public class SingleNodeMessageConsumerService {

    /**
     * 监听一个主题
     * @param records
     * @param ack
     */
    @KafkaListener(id = "haha",groupId = "sadan",topics = "llc2021",clientIdPrefix = "orion",concurrency = "3")
    public void singleConsume01(List<ConsumerRecord<String, Object>> records, Acknowledgment ack){
        System.out.println("--------- sadan group 接收并处理消息 ---------");
        for (ConsumerRecord<String, Object> record : records) {
            System.out.println(Thread.currentThread()+"-----消费者收到消息:" + record + "; ack:" + ack);
        }
        ack.acknowledge();
    }


    /**
     * 监听多个主题,设置过多的concurrency会浪费资源，比如这里2个topic都少于5个分区的，会有一个消费者获取不到partition消费。
     * @param records
     * @param ack
     */
    @KafkaListener(groupId = "sadan_fake",topics = {"llc2021","llc2022"},concurrency = "5")
    public void singleConsume02(List<ConsumerRecord<String, Object>> records, Acknowledgment ack){
        System.out.println("--------- sadan_fake group  接收并处理消息 ---------");
        for (ConsumerRecord<String, Object> record : records) {
            System.out.println(Thread.currentThread()+"-----消费者收到消息:" + record + "; ack:" + ack);
        }
        ack.acknowledge();
    }
}
